package com.example.flinkcourse.lesson1.time;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;

/**
 * 时间语义处理类
 * 实现事件时间处理和水位线机制
 */
public class TimeAssigner implements WatermarkStrategy<String> {
    private static final Logger LOG = LoggerFactory.getLogger(TimeAssigner.class);
    private static final long MAX_OUT_OF_ORDERNESS = 5000; // 5 seconds

    @Override
    public TimestampAssigner<String> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return new TimestampAssigner<String>() {
            @Override
            public long extractTimestamp(String element, long recordTimestamp) {
                // 示例：从数据中提取时间戳
                // 这里假设数据格式为 "timestamp:content"
                try {
                    String[] parts = element.split(":", 2);
                    if (parts.length == 2) {
                        return Long.parseLong(parts[0]);
                    }
                } catch (Exception e) {
                    LOG.warn("Failed to extract timestamp from: {}", element, e);
                }
                return System.currentTimeMillis();
            }
        };
    }

    @Override
    public WatermarkGenerator<String> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new WatermarkGenerator<String>() {
            private long maxTimestamp;

            @Override
            public void onEvent(String event, long eventTimestamp, WatermarkOutput output) {
                maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                output.emitWatermark(new org.apache.flink.api.common.eventtime.Watermark(
                    maxTimestamp - MAX_OUT_OF_ORDERNESS));
            }
        };
    }

    public static WatermarkStrategy<String> withWatermarks() {
        return new TimeAssigner()
            .withTimestampAssigner((event, timestamp) -> {
                try {
                    String[] parts = event.split(":", 2);
                    if (parts.length == 2) {
                        return Long.parseLong(parts[0]);
                    }
                } catch (Exception e) {
                    LOG.warn("Failed to extract timestamp from: {}", event, e);
                }
                return System.currentTimeMillis();
            })
            .withIdleness(Duration.ofSeconds(1))
            .withBoundedOutOfOrderness(Duration.ofMillis(MAX_OUT_OF_ORDERNESS));
    }
} 